#include "Win32AsynchIO.h"
#include "MessageBlock.h"
#include "Win32Proactor.h"
#include "Debug.h"
#include "Trace.h"

Win32AsynchOpt::Win32AsynchOpt (Win32Proactor *ep)
: win32_proactor_ (ep)
, proactor_ (0)
, asynch_handler_ (0)			      
, handle_ (NDK_INVALID_HANDLE)
, result_ (0)
{
    TRACE ("");
}
Win32AsynchOpt::~Win32AsynchOpt ()
{
    TRACE ("");
    if (this->result_)
    {
	delete this->result_;
	this->result_ = 0;
    }
}
int Win32AsynchOpt::open (const AsynchHandler *handler,
	NDK_HANDLE handle,
	const void * /* completion_key */,
	Proactor *proactor)
{
    TRACE ("");
    if (handler == 0) return -1;

    this->proactor_ = proactor;
    this->asynch_handler_ = const_cast<AsynchHandler *>(handler);
    this->handle_ = handle;

    if (this->handle_ == NDK_INVALID_HANDLE)
	this->handle_ = handler->handle ();

    if (this->handle_ == NDK_INVALID_HANDLE)
	return -1;

    return 0;
}
// ---------------------------------------------------------------------
Win32AsynchResult::Win32AsynchResult (const AsynchHandler *handler,
	NDK_HANDLE handle,
	long long offset)
: handle_ (handle)
, bytes_transferred_ (0)
, success_ (0)
, completion_key_ (0)
, asynch_handler_ (const_cast<AsynchHandler*>(handler))
, error_ (0)
, offset_ (offset)
{
    TRACE ("");
}
Win32AsynchResult::~Win32AsynchResult ()
{
    TRACE ("");
}
// ---------------------------------------------------------------------
Win32AsynchReadStream::Win32AsynchReadStream (Win32Proactor *p)
: AsynchReadStreamImpl () 
, Win32AsynchOpt (p)
{
    TRACE ("");
}
Win32AsynchReadStream::~Win32AsynchReadStream ()
{
    TRACE ("");
}
int Win32AsynchReadStream::open (const AsynchHandler *handler,
	NDK_HANDLE handle,
	const void *completion_key,
	Proactor *proactor/* = 0*/)
{
    TRACE ("");
    if (Win32AsynchOpt::open (handler,
		handle, 
		completion_key,
		proactor) == 0)
    {
	this->result_ = new Win32AsynchReadStreamResult(
		this->asynch_handler_,
		this->handle_,
		0,
		0);
	return 0;
    }
    return -1;
}
int Win32AsynchReadStream::cancel (void)
{
    TRACE ("");
    if (!this->win32_proactor_)
	return -1;

    return this->win32_proactor_->cancel_aio (this->handle_,
	    AsynchHandler::READ_MASK);
}
int Win32AsynchReadStream::read (MessageBlock *message_block,
	size_t bytes_to_read)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (bytes_to_read > message_block->size ())
	bytes_to_read = message_block->size ();
    if (bytes_to_read == 0)
	return -1;
    if (this->result_ == 0) return -1;
    /*
    Win32AsynchReadStreamResult *result = new Win32AsynchReadStreamResult(
	    this->asynch_handler_,
	    this->handle_,
	    message_block,
	    bytes_to_read);
	    */
    Win32AsynchReadStreamResult *r 
	= dynamic_cast<Win32AsynchReadStreamResult *>(this->result_);
    r->message_block (message_block);
    r->bytes_to_read (bytes_to_read);

    WSABUF buf = 
    {
	(u_long)bytes_to_read, 
	message_block->wr_ptr ()
    };
    DWORD bytes_recvd;
    DWORD flag = 0;
    
    int result = ::WSARecv(this->handle_, 
	    &buf,
	    1,
	    &bytes_recvd,
	    &flag,
	    this->result_,
	    0);
    if (result != 0 && GetLastError()!= WSA_IO_PENDING)
    {
	return -1;
    }
    return 0;
}
Win32AsynchReadStreamResult::Win32AsynchReadStreamResult (
	const AsynchHandler *handler,
	NDK_HANDLE handle,
	MessageBlock *message_block,
	size_t bytes_to_read)
: AsynchReadStreamResultImpl ()
, Win32AsynchResult (handler, handle, 0)
, bytes_to_read_ (bytes_to_read)
, message_block_ (message_block)
{
    TRACE ("");
}
Win32AsynchReadStreamResult::~Win32AsynchReadStreamResult ()
{
    TRACE ("");
}
// ---------------------------------------------------------------------
Win32AsynchWriteStream::Win32AsynchWriteStream (Win32Proactor *p)
: AsynchWriteStreamImpl () 
, Win32AsynchOpt (p)
{
    TRACE ("");
}
Win32AsynchWriteStream::~Win32AsynchWriteStream ()
{
    TRACE ("");
}
int Win32AsynchWriteStream::open (const AsynchHandler *handler,
	NDK_HANDLE handle,
	const void * completion_key,
	Proactor *proactor)
{
    TRACE ("");
    if (Win32AsynchOpt::open (handler,
		handle,
		completion_key,
		proactor) == 0)
    {
	this->result_ = new Win32AsynchWriteStreamResult(
		this->asynch_handler_,
		this->handle_,
		0,
		0);
	return 0;
    }
    return -1;
}
int Win32AsynchWriteStream::cancel (void)
{
    TRACE ("");
    if (!this->win32_proactor_)
	return -1;

    return this->win32_proactor_->cancel_aio (this->handle_,
	    AsynchHandler::WRITE_MASK);
}
int Win32AsynchWriteStream::write (MessageBlock *message_block,
	size_t bytes_to_write)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (bytes_to_write > message_block->length ())
	bytes_to_write = message_block->length ();
    if (bytes_to_write == 0)
	return -1;
    if (this->result_ == 0) return -1;
    /*
    Win32AsynchWriteStreamResult *result = new Win32AsynchWriteStreamResult(
	    this->asynch_handler_,
	    this->handle_,
	    message_block,
	    bytes_to_write);
	    */
    Win32AsynchWriteStreamResult *r 
	= dynamic_cast<Win32AsynchWriteStreamResult *>(this->result_);
    r->message_block (message_block);
    r->bytes_to_write (bytes_to_write);
    return this->proactor_impl ()->start_aio (this, 
	    AsynchHandler::WRITE_MASK);
}
Win32AsynchWriteStreamResult::Win32AsynchWriteStreamResult (
	const AsynchHandler *handler,
	NDK_HANDLE handle,
	MessageBlock *message_block,
	size_t bytes_to_write)
: AsynchWriteStreamResultImpl ()
, Win32AsynchResult (handler, handle, 0)
, bytes_to_write_ (bytes_to_write)
, message_block_ (message_block)
{
    TRACE ("");
}
Win32AsynchWriteStreamResult::~Win32AsynchWriteStreamResult ()
{
}
// ---------------------------------------------------------------------
Win32AsynchAccept::Win32AsynchAccept (Win32Proactor *p)
: AsynchAcceptImpl ()
, Win32AsynchOpt (p)
{
    TRACE ("");
}
Win32AsynchAccept::~Win32AsynchAccept ()
{
    TRACE ("");
}
int Win32AsynchAccept::open (const AsynchHandler *handler,
	NDK_HANDLE handle,
	const void *completion_key,
	Proactor *proactor)
{
    TRACE ("");
    if (Win32AsynchOpt::open (handler,
		handle,
		completion_key,
		proactor) == 0)
    {
	this->result_ = new Win32AsynchAcceptResult(
		this->asynch_handler_,
		this->handle_,
		0,
		0);
	return 0;
    }
    return -1;
}
int Win32AsynchAccept::cancel (void)
{
    TRACE ("");
    if (!this->win32_proactor_)
	return -1;

    return this->win32_proactor_->cancel_aio (this->handle_,
	    AsynchHandler::READ_MASK);
}
int Win32AsynchAccept::accept (MessageBlock *message_block,
	size_t num_to_accept)
{
    TRACE ("");
    if (message_block == 0) return -1;
    if (sizeof (ConnectionInfo) * num_to_accept > message_block->size ())
    {
	num_to_accept = message_block->size () / sizeof (ConnectionInfo);
    }
    if (num_to_accept == 0)
	return -1;
    if (this->result_ == 0) return -1;

    /*
    Win32AsynchAcceptResult *result = new Win32AsynchAcceptResult(
	    this->asynch_handler_,
	    message_block,
	    this->handle_,
	    num_to_accept);
	    */
    Win32AsynchAcceptResult *r = 
	dynamic_cast<Win32AsynchAcceptResult *>(this->result_);
    r->message_block (message_block);
    r->num_to_accept (num_to_accept);
    return this->proactor_impl ()->start_aio (this, 
	    AsynchHandler::READ_MASK);
}
int Win32AsynchAccept::execute (Win32AsynchResult *&apt_result)
{
    TRACE ("");
    int finished = 0;
    size_t num_accepted = 0;
    NDK_HANDLE accept_handle = NDK_INVALID_HANDLE;

    apt_result = this->result_;
    Win32AsynchAcceptResult *accept_result =
	dynamic_cast<Win32AsynchAcceptResult *>(this->result_);
    accept_result->success_ = 1;
    accept_result->error_ = 0;   // refresh;

    while(num_accepted < accept_result->num_to_accept_)
    {
	InetAddr remote_addr;
	sockaddr *addr   = 0;
	addr    = (sockaddr *) remote_addr.get_addr ();
	int len = remote_addr.get_addr_size ();
	accept_handle = NDK::accept (accept_result->listen_handle_,
		addr,
		&len);
	if (accept_handle == -1)
	{
	    if (errno == EAGAIN)  // special for Edge Triggered model.
	    {
		accept_result->error_ = EAGAIN;
		finished = 1;
		break;
	    }else if (errno == EINTR)
	    {
		continue;
	    }else  // error
	    {
		accept_result->error_ = errno;
		accept_result->success_ = 0;
		finished = 1;
		break;
	    }
	}
	//NDK_INF ("recv a connection %d, listen handle = %d\n", 
		//accept_handle,
		//accept_result->listen_handle_);
	ConnectionInfo info (remote_addr, accept_handle);
	::memcpy (accept_result->message_block_->wr_ptr (), 
		&info, 
		sizeof (info));
	accept_result->message_block_->wr_ptr (sizeof (info));
	++num_accepted;
    }
    return finished;
}
Win32AsynchAcceptResult::Win32AsynchAcceptResult (
	const AsynchHandler *handler,
	NDK_HANDLE listen_handle,
	MessageBlock *message_block,
	size_t num_to_accept)
: AsynchAcceptResultImpl ()
, Win32AsynchResult (handler, listen_handle, 0)
, listen_handle_ (listen_handle)
, num_to_accept_ (num_to_accept)
, message_block_ (message_block)
{
    TRACE ("");
}
Win32AsynchAcceptResult::~Win32AsynchAcceptResult ()
{
    TRACE ("");
}


